package org.zyj.consumer;

import com.alibaba.fastjson.JSONObject;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.MessageProperties;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;
import org.zyj.dto.MessageDto;

import javax.annotation.Resource;
import java.io.IOException;

@Component
public class ConsumerTest {
    @Resource
    private RedisTemplate redisTemplate;

    /**
     * 消费者1
     */
    @RabbitListener(queues = "test01")
    public void consumer1(String msg, Message message, Channel channel) {
        System.out.println("消费者----1 接收到消息=" + msg);
        MessageDto dto = JSONObject.parseObject(msg, MessageDto.class);
        if (dto == null) return;
        int id = dto.getId();
        if (id == 1) {
            //说明是第一条消息，那就可以直接处理
            System.out.println("消费者----1 处理消息id=" + id);
            //已处理的消息ID，存到Redis中
            redisTemplate.opsForValue().set(String.valueOf(id), id);
            try {
                //这里模拟业务代码处理耗时1.5秒
                System.out.println("----------");
                Thread.sleep(1500);
                //业务代码处理完毕后，进行手动ACK,false:删除队列中的当前消息
                channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
            } catch (InterruptedException | IOException e) {
                e.printStackTrace();
            }
        } else {
            //说明不是第一条消息，那就需要验证前一条消息是否已被成功消费了（就是看是否存在对应key的值，存在：表明已处理，不存在：未处理）
            String redisKey = String.valueOf(id - 1);
            Object redisValue = redisTemplate.opsForValue().get(redisKey);
            if (redisValue != null) {
                //说明前一条消息已经处理完毕了，那就可以开始处理这条消息了
                System.out.println("消费者****2 处理消息id=" + id);
                //已处理的消息ID，存到Redis中
                redisTemplate.opsForValue().set(String.valueOf(id), id);
                try {
                    //这里模拟业务代码处理耗时1.5秒
                    Thread.sleep(1500);
                    //业务代码处理完毕后，进行手动ACK,false:删除队列中的当前消息
                    channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
                    //消息处理完毕后，删除前一条消息的key，以免产生垃圾数据
                    redisTemplate.delete(redisKey);
                } catch (InterruptedException | IOException e) {
                    e.printStackTrace();
                }
            } else {
                //此时说明前一条消息还未消费，那么就需要重回队列了
                try {
                    /*
                     * 第二个参数含义：如果被拒绝的消息应该重新排队，而不是丢弃掉，则为true
                     * 经过测试发现 basicReject 方法并不会将消息放到队尾，而是队列头部，所
                     * 以我们不能使用 basicReject 方法
                     * 因此我们采取的解决方案是：
                     * 将消息进行应答，这时消息队列会删除该消息，同时我们再次发送该消息到消
                     息队列，这时就实现了错误消息进行消息队列尾部的效果
                     */
                    //channel.basicReject(message.getMessageProperties().getDeliveryTag(), true);
                    //先进行手动ACK，删除RabbitMQ中的消息
                    channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
                    //然后再将消息添加到队列的尾部
                    channel.basicPublish(message.getMessageProperties().getReceivedExchange(),
                            message.getMessageProperties().getReceivedRoutingKey(),
                            MessageProperties.PERSISTENT_TEXT_PLAIN,
                            message.getBody());
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    }

    /**
     * 消费者2
     */
    @RabbitListener(queues = "test01")
    public void consumer2(String msg, Message message, Channel channel) {
        System.out.println("消费者****2 接收到消息=" + msg);
        MessageDto dto = JSONObject.parseObject(msg, MessageDto.class);
        if (dto == null) return;
        int id = dto.getId();
        if (id == 1) {
            //说明是第一条消息，那就可以直接处理
            System.out.println("消费者****2 处理消息id=" + id);
            //已处理的消息ID，存到Redis中
            redisTemplate.opsForValue().set(String.valueOf(id), id);
            try {
                //这里模拟业务代码处理耗时1.5秒
                Thread.sleep(1500);
                //业务代码处理完毕后，进行手动ACK
                channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
            } catch (InterruptedException | IOException e) {
                e.printStackTrace();
            }
        } else {
            //说明不是第一条消息，那就需要验证前一条消息是否已被成功消费了
            String redisKey = String.valueOf(id - 1);
            Object redisValue = redisTemplate.opsForValue().get(redisKey);
            if (redisValue != null) {
                //说明前一条消息已经处理完毕了，那就可以开始处理这条消息了
                System.out.println("消费者----1 处理消息id=" + id);
                //已处理的消息ID，存到Redis中
                redisTemplate.opsForValue().set(String.valueOf(id), id);
                try {
                    //这里模拟业务代码处理耗时1.5秒
                    Thread.sleep(1500);
                    //业务代码处理完毕后，进行手动ACK
                    channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
                    //消息处理完毕后，删除前一条消息的key，以免产生垃圾数据
                    redisTemplate.delete(redisKey);
                } catch (InterruptedException | IOException e) {
                    e.printStackTrace();
                }
            } else {
                //此时说明前一条消息还未消费，那么就需要重回队列了
                try {
                    /*
                     * 第二个参数含义：如果被拒绝的消息应该重新排队，而不是丢弃掉，则为true
                     * 经过测试发现 basicReject 方法并不会将消息放到队尾，而是队列头部，所
                     * 以我们不能使用 basicReject 方法
                     * 因此我们采取的解决方案是：
                     * 将消息进行应答，这时消息队列会删除该消息，同时我们再次发送该消息到消
                     * 息队列，这时就实现了错误消息进行消息队列尾部的效果
                     */
                    //channel.basicReject(message.getMessageProperties().getDeliveryTag(), true);
                    //先进行手动ACK，删除RabbitMQ中的消息
                    channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
                    //然后再将消息添加到队列的尾部
                    channel.basicPublish(message.getMessageProperties().getReceivedExchange(),
                            message.getMessageProperties().getReceivedRoutingKey(),
                            MessageProperties.PERSISTENT_TEXT_PLAIN,
                            message.getBody());
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    }
}